<!DOCTYPE html>
<html lang="zh-cn">
<head>
    
    <link type="text/css" rel="stylesheet" href="/bundles/blog-common.css?v=KOZafwuaDasEedEenI5aTy8aXH0epbm6VUJ0v3vsT_Q1"/>
<link id="MainCss" type="text/css" rel="stylesheet" href="/skins/ThinkInside/bundle-ThinkInside.css?v=RRjf6pEarGnbXZ86qxNycPfQivwSKWRa4heYLB15rVE1"/>
<link type="text/css" rel="stylesheet" href="/blog/customcss/428549.css?v=%2fam3bBTkW5NBWhBE%2fD0lcyJv5UM%3d"/>

</head>
<body>
<a name="top"></a>

<div id="page_begin_html"></div><script>load_page_begin_html();</script>

<div id="topics">
	<div class = "post">
		<h1 class = "postTitle">
			<a id="cb_post_title_url" class="postTitle2" href="https://www.cnblogs.com/frankdeng/p/9301672.html">Spark（四）Spark之Transformation和Action</a>
		</h1>
		<div class="clear"></div>
		<div class="postBody">
			<div id="cnblogs_post_body" class="blogpost-body"><h2>Transformation算子</h2>
<h3>基本的初始化</h3>
<h4>java</h4>
<div class="cnblogs_code">
<pre>static SparkConf conf = null;
static JavaSparkContext sc = null;
static {
     conf = new SparkConf();
     conf.setMaster("local").setAppName("TestTransformation");
     sc = new JavaSparkContext(conf);
}</pre>
</div>
<h4>scala</h4>
<div class="cnblogs_code">
<pre>private val conf: SparkConf = new SparkConf().setAppName("TestTransformation").setMaster("local")
private val sparkContext = new SparkContext(conf)</pre>
</div>
<h2>一、map、flatMap、mapParations、mapPartitionsWithIndex</h2>
<h3>1.1　map</h3>
<h4>（1）　使用Java7进行编写</h4>
<p>map十分容易理解，他是将源JavaRDD的一个一个元素的传入call方法，并经过算法后一个一个的返回从而生成一个新的JavaRDD。&nbsp;</p>
<div class="cnblogs_code">
<pre>public static void map(){
        //String[] names = {"张无忌","赵敏","周芷若"};
        List&lt;String&gt; list = Arrays.asList("张无忌","赵敏","周芷若");
        System.out.println(list.size());
        JavaRDD&lt;String&gt; listRDD = sc.parallelize(list);

        JavaRDD&lt;String&gt; nameRDD = listRDD.map(new Function&lt;String, String&gt;() {
            @Override
            public String call(String name) throws Exception {
                return "Hello " + name;
            }
        });

        nameRDD.foreach(new VoidFunction&lt;String&gt;() {
            @Override
            public void call(String s) throws Exception {
                System.out.println(s);
            }
        });
    }</pre>
</div>
<h4>（2）　使用Java8编写</h4>
<div class="cnblogs_code">
<pre>public static void map(){
        String[] names = {"张无忌","赵敏","周芷若"};
        List&lt;String&gt; list = Arrays.asList(names);
        JavaRDD&lt;String&gt; listRDD = sc.parallelize(list);

        JavaRDD&lt;String&gt; nameRDD = listRDD.map(name -&gt; {
            return "Hello " + name;
        });

        nameRDD.foreach(name -&gt; System.out.println(name));

    }</pre>
</div>
<h4>（3）　使用scala进行编写</h4>
<div class="cnblogs_code">
<pre>def map(): Unit ={
    val list = List("张无忌", "赵敏", "周芷若")
    val listRDD = sc.parallelize(list)
    val nameRDD = listRDD.map(name =&gt; "Hello " + name)
    nameRDD.foreach(name =&gt; println(name))
  }</pre>
</div>
<h4>（4）　运行结果</h4>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201804/1228818-20180425145949797-1293325450.png" alt="" /></p>
<h4>（5）　总结</h4>
<p>可以看出，对于map算子，源JavaRDD的每个元素都会进行计算，由于是依次进行传参，所以他是有序的，新RDD的元素顺序与源RDD是相同的。而由有序又引出接下来的flatMap。</p>
<h3>1.2　flatMap</h3>
<h4>（1）　使用Java7进行编写</h4>
<p>flatMap与map一样，是将RDD中的元素依次的传入call方法，他比map多的功能是能在任何一个传入call方法的元素后面添加任意多元素，而能达到这一点，正是因为其进行传参是依次进行的。</p>
<div class="cnblogs_code">
<pre> public static void flatMap(){
        List&lt;String&gt; list = Arrays.asList("张无忌 赵敏","宋青书 周芷若");
        JavaRDD&lt;String&gt; listRDD = sc.parallelize(list);

        JavaRDD&lt;String&gt; nameRDD = listRDD <br />                 .flatMap(new FlatMapFunction&lt;String, String&gt;() {
            @Override
            public Iterator&lt;String&gt; call(String line) throws Exception {
                return Arrays.asList(line.split(" ")).iterator();
            }
        })
                .map(new Function&lt;String, String&gt;() {
                    @Override
                    public String call(String name) throws Exception {
                        return "Hello " + name;
                    }
                });

        nameRDD.foreach(new VoidFunction&lt;String&gt;() {
            @Override
            public void call(String s) throws Exception {
                System.out.println(s);
            }
        });

    }</pre>
</div>
<h4>（2）　使用Java8进行编写</h4>
<div class="cnblogs_code">
<pre>public static void flatMap(){
        List&lt;String&gt; list = Arrays.asList("张无忌 赵敏","宋青书 周芷若");
        JavaRDD&lt;String&gt; listRDD = sc.parallelize(list);

        JavaRDD&lt;String&gt; nameRDD = listRDD.flatMap(line -&gt; Arrays.asList(line.split(" ")).iterator()).map(name -&gt; "Hello " + name);

        nameRDD.foreach(name -&gt; System.out.println(name));
    }</pre>
</div>
<h4>（3）　使用scala进行编写</h4>
<div class="cnblogs_code">
<pre>def flatMap(): Unit ={
    val list = List("张无忌 赵敏","宋青书 周芷若")
    val listRDD = sc.parallelize(list)

    val nameRDD = listRDD.flatMap(line =&gt; line.split(" ")).map(name =&gt; "Hello " + name)
    nameRDD.foreach(name =&gt; println(name))
  }</pre>
</div>
<h4>（4）　运行结果</h4>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201804/1228818-20180423203536866-68590234.png" alt="" /></p>
<h4>（5）　总结</h4>
<p>flatMap的特性决定了这个算子在对需要随时增加元素的时候十分好用，比如在对源RDD查漏补缺时。</p>
<p>map和flatMap都是依次进行参数传递的，但有时候需要RDD中的两个元素进行相应操作时（例如：算存款所得时，下一个月所得的利息是要原本金加上上一个月所得的本金的），这两个算子便无法达到目的了，这是便需要mapPartitions算子，他传参的方式是将整个RDD传入，然后将一个迭代器传出生成一个新的RDD，由于整个RDD都传入了，所以便能完成前面说的业务。</p>
<h3>1.3　mapPartitions</h3>
<h4>（1）　使用Java7进行编写</h4>
<div class="cnblogs_code">
<pre>/**
     * map:
     *    一条数据一条数据的处理（文件系统，数据库等等）
     * mapPartitions：
     *    一次获取的是一个分区的数据（hdfs）
     *    正常情况下，mapPartitions 是一个高性能的算子
     *    因为每次处理的是一个分区的数据，减少了去获取数据的次数。
     *
     *    但是如果我们的分区如果设置得不合理，有可能导致每个分区里面的数据量过大。
     */
    public static void mapPartitions(){
        List&lt;Integer&gt; list = Arrays.asList(1, 2, 3, 4, 5, 6);
        //参数二代表这个rdd里面有两个分区
        JavaRDD&lt;Integer&gt; listRDD = sc.parallelize(list,2);

        listRDD.mapPartitions(new FlatMapFunction&lt;Iterator&lt;Integer&gt;, String&gt;() {
            @Override
            public Iterator&lt;String&gt; call(Iterator&lt;Integer&gt; iterator) throws Exception {
                ArrayList&lt;String&gt; array = new ArrayList&lt;&gt;();
                while (iterator.hasNext()){
                    array.add("hello " + iterator.next());
                }
                return array.iterator();
            }
        }).foreach(new VoidFunction&lt;String&gt;() {
            @Override
            public void call(String s) throws Exception {
                System.out.println(s);
            }
        });
    }</pre>
</div>
<h4>（2）　使用Java8进行编写</h4>
<div class="cnblogs_code">
<pre>public static void mapParations(){
        List&lt;Integer&gt; list = Arrays.asList(1, 2, 3, 4, 5, 6);
        JavaRDD&lt;Integer&gt; listRDD = sc.parallelize(list, 2);

        listRDD.mapPartitions(iterator -&gt; {
            ArrayList&lt;String&gt; array = new ArrayList&lt;&gt;();
            while (iterator.hasNext()){
                array.add("hello " + iterator.next());
            }
            return array.iterator();
        }).foreach(name -&gt; System.out.println(name));
    }</pre>
</div>
<h4>（3）　使用scala进行编写</h4>
<div class="cnblogs_code">
<pre>  def mapParations(): Unit ={
    val list = List(1,2,3,4,5,6)
    val listRDD = sc.parallelize(list,2)

    listRDD.mapPartitions(iterator =&gt; {
      val newList: ListBuffer[String] = ListBuffer()
      while (iterator.hasNext){
        newList.append("hello " + iterator.next())
      }
      newList.toIterator
    }).foreach(name =&gt; println(name))
  }</pre>
</div>
<h4>（4）　运行结果</h4>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201804/1228818-20180423205910060-1126297392.png" alt="" /></p>
<h3>1.4　mapPartitionsWithIndex</h3>
<p>每次获取和处理的就是一个分区的数据,并且知道处理的分区的分区号是啥？</p>
<h4>（1）使用Java7编写</h4>
<div class="cnblogs_code">
<pre>public static void mapPartitionsWithIndex(){
        List&lt;Integer&gt; list = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8);
        JavaRDD&lt;Integer&gt; listRDD = sc.parallelize(list, 2);
        listRDD.mapPartitionsWithIndex(new Function2&lt;Integer, Iterator&lt;Integer&gt;, Iterator&lt;String&gt;&gt;() {
            @Override
            public Iterator&lt;String&gt; call(Integer index, Iterator&lt;Integer&gt; iterator) throws Exception {
                ArrayList&lt;String&gt; list1 = new ArrayList&lt;&gt;();
                while (iterator.hasNext()){
                    list1.add(index+"_"+iterator.next());
                }
                return list1.iterator();
            }
        },true)
                .foreach(new VoidFunction&lt;String&gt;() {
            @Override
            public void call(String s) throws Exception {
                System.out.println(s);
            }
        });
    }</pre>
</div>
<h4>（2）使用Java8编写</h4>
<div class="cnblogs_code">
<pre> public static void mapPartitionsWithIndex() {
        List&lt;Integer&gt; list = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8);
        JavaRDD&lt;Integer&gt; listRDD = sc.parallelize(list, 2);
        listRDD.mapPartitionsWithIndex((index,iterator) -&gt; {
            ArrayList&lt;String&gt; list1 = new ArrayList&lt;&gt;();
            while (iterator.hasNext()){
                list1.add(index+"_"+iterator.next());
            }
            return list1.iterator();
        },true)
                .foreach(str -&gt; System.out.println(str));
    }</pre>
</div>
<h4>（3）使用scala编写</h4>
<div class="cnblogs_code">
<pre>def mapPartitionsWithIndex(): Unit ={
    val list = List(1,2,3,4,5,6,7,8)
    sc.parallelize(list).mapPartitionsWithIndex((index,iterator) =&gt; {
      val listBuffer:ListBuffer[String] = new ListBuffer
      while (iterator.hasNext){
        listBuffer.append(index+"_"+iterator.next())
      }
      listBuffer.iterator
    },true)
      .foreach(println(_))
  }</pre>
</div>
<h4>（4）运行结果</h4>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201804/1228818-20180424211736405-1885495223.png" alt="" /></p>
<h2>二、reduce、reduceByKey</h2>
<h3>2.1　reduce</h3>
<p>reduce其实是将RDD中的所有元素进行合并，当运行call方法时，会传入两个参数，在call方法中将两个参数合并后返回，而这个返回值会和一个新的RDD中的元素再次传入call方法中，继续合并，直到合并到只剩下一个元素时。</p>
<h4>（1）使用Java7编写</h4>
<div class="cnblogs_code">
<pre>public static void reduce(){
        List&lt;Integer&gt; list = Arrays.asList(1, 2, 3, 4, 5, 6);
        JavaRDD&lt;Integer&gt; listRDD = sc.parallelize(list);

        Integer result = listRDD.reduce(new Function2&lt;Integer, Integer, Integer&gt;() {
            @Override
            public Integer call(Integer i1, Integer i2) throws Exception {
                return i1 + i2;
            }
        });
        System.out.println(result);

    }</pre>
</div>
<h4>（2）使用Java8编写</h4>
<div class="cnblogs_code">
<pre>public static void reduce(){
        List&lt;Integer&gt; list = Arrays.asList(1, 2, 3, 4, 5, 6);
        JavaRDD&lt;Integer&gt; listRDD = sc.parallelize(list);

        Integer result = listRDD.reduce((x, y) -&gt; x + y);
        System.out.println(result);
    }</pre>
</div>
<h4>（3）使用scala编写</h4>
<div class="cnblogs_code">
<pre> def reduce(): Unit ={
    val list = List(1,2,3,4,5,6)
    val listRDD = sc.parallelize(list)
    
    val result = listRDD.reduce((x,y) =&gt; x+y)
    println(result)
  }</pre>
</div>
<h4>（4）运行结果</h4>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201804/1228818-20180423211543840-599401739.png" alt="" /></p>
<h3>2.2　reduceByKey</h3>
<p>reduceByKey仅将RDD中所有K,V对中K值相同的V进行合并。</p>
<h4>（1）使用Java7编写</h4>
<div class="cnblogs_code">
<pre>public static void reduceByKey(){
        List&lt;Tuple2&lt;String, Integer&gt;&gt; list = Arrays.asList(
                new Tuple2&lt;String, Integer&gt;("武当", 99),
                new Tuple2&lt;String, Integer&gt;("少林", 97),
                new Tuple2&lt;String, Integer&gt;("武当", 89),
                new Tuple2&lt;String, Integer&gt;("少林", 77)
        );
        JavaPairRDD&lt;String, Integer&gt; listRDD = sc.parallelizePairs(list);
        //运行reduceByKey时，会将key值相同的组合在一起做call方法中的操作
        JavaPairRDD&lt;String, Integer&gt; result = listRDD.reduceByKey(new Function2&lt;Integer, Integer, Integer&gt;() {
            @Override
            public Integer call(Integer i1, Integer i2) throws Exception {
                return i1 + i2;
            }
        });
        result.foreach(new VoidFunction&lt;Tuple2&lt;String, Integer&gt;&gt;() {
            @Override
            public void call(Tuple2&lt;String, Integer&gt; tuple) throws Exception {
                System.out.println("门派: " + tuple._1 + "-&gt;" + tuple._2);
            }
        });
    }</pre>
</div>
<h4>（2）使用Java8编写</h4>
<div class="cnblogs_code">
<pre>public static void reduceByKey(){
        List&lt;Tuple2&lt;String, Integer&gt;&gt; list = Arrays.asList(
                new Tuple2&lt;String, Integer&gt;("武当", 99),
                new Tuple2&lt;String, Integer&gt;("少林", 97),
                new Tuple2&lt;String, Integer&gt;("武当", 89),
                new Tuple2&lt;String, Integer&gt;("少林", 77)
        );
        JavaPairRDD&lt;String, Integer&gt; listRDD = sc.parallelizePairs(list);

        JavaPairRDD&lt;String, Integer&gt; resultRDD = listRDD.reduceByKey((x, y) -&gt; x + y);
        resultRDD.foreach(tuple -&gt; System.out.println("门派: " + tuple._1 + "-&gt;" + tuple._2));
    }</pre>
</div>
<h4>（3）使用scala编写</h4>
<div class="cnblogs_code">
<pre>def reduceByKey(): Unit ={
    val list = List(("武当", 99), ("少林", 97), ("武当", 89), ("少林", 77))
    val mapRDD = sc.parallelize(list)

    val resultRDD = mapRDD.reduceByKey(_+_)
    resultRDD.foreach(tuple =&gt; println("门派: " + tuple._1 + "-&gt;" + tuple._2))
  }</pre>
</div>
<h4>（4）运行结果</h4>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201804/1228818-20180423215428805-1389370640.png" alt="" /></p>
<h2>三、union，join和groupByKey&nbsp;</h2>
<h3>3.1　union</h3>
<p>当要将两个RDD合并时，便要用到union和join，其中union只是简单的将两个RDD累加起来，可以看做List的addAll方法。就想List中一样，当使用union及join时，必须保证两个RDD的泛型是一致的。</p>
<h4>（1）使用Java7编写</h4>
<div class="cnblogs_code">
<pre>public static void union(){
        final List&lt;Integer&gt; list1 = Arrays.asList(1, 2, 3, 4);
        final List&lt;Integer&gt; list2 = Arrays.asList(3, 4, 5, 6);
        final JavaRDD&lt;Integer&gt; rdd1 = sc.parallelize(list1);
        final JavaRDD&lt;Integer&gt; rdd2 = sc.parallelize(list2);
        rdd1.union(rdd2)
                .foreach(new VoidFunction&lt;Integer&gt;() {
                    @Override
                    public void call(Integer number) throws Exception {
                        System.out.println(number + "");
                    }
                });
    }</pre>
</div>
<h4>（2）使用Java8编写</h4>
<div class="cnblogs_code">
<pre>public static void union(){
        final List&lt;Integer&gt; list1 = Arrays.asList(1, 2, 3, 4);
        final List&lt;Integer&gt; list2 = Arrays.asList(3, 4, 5, 6);
        final JavaRDD&lt;Integer&gt; rdd1 = sc.parallelize(list1);
        final JavaRDD&lt;Integer&gt; rdd2 = sc.parallelize(list2);

        rdd1.union(rdd2).foreach(num -&gt; System.out.println(num));
    }</pre>
</div>
<h4>（3）使用scala编写</h4>
<div class="cnblogs_code">
<pre>def union(): Unit ={
    val list1 = List(1,2,3,4)
    val list2 = List(3,4,5,6)
    val rdd1 = sc.parallelize(list1)
    val rdd2 = sc.parallelize(list2)
    rdd1.union(rdd2).foreach(println(_))
  }</pre>
</div>
<h4>（4）运行结果</h4>
<p>&nbsp;<img src="https://images2018.cnblogs.com/blog/1228818/201804/1228818-20180423220431875-1286219614.png" alt="" /></p>
<h3>3.2　groupByKey</h3>
<h4>（1）使用Java7编写</h4>
<p>union只是将两个RDD简单的累加在一起，而join则不一样，join类似于hadoop中的combin操作，只是少了排序这一段，再说join之前说说groupByKey，因为join可以理解为union与groupByKey的结合：groupBy是将RDD中的元素进行分组，组名是call方法中的返回值，而顾名思义groupByKey是将PairRDD中拥有相同key值得元素归为一组。即：</p>
<div class="cnblogs_code">
<pre>public static void groupByKey(){
        List&lt;Tuple2&lt;String,String&gt;&gt; list = Arrays.asList(
                new Tuple2("武当", "张三丰"),
                new Tuple2("峨眉", "灭绝师太"),
                new Tuple2("武当", "宋青书"),
                new Tuple2("峨眉", "周芷若")
        );
        JavaPairRDD&lt;String, String&gt; listRDD = sc.parallelizePairs(list);

        JavaPairRDD&lt;String, Iterable&lt;String&gt;&gt; groupByKeyRDD = listRDD.groupByKey();
        groupByKeyRDD.foreach(new VoidFunction&lt;Tuple2&lt;String, Iterable&lt;String&gt;&gt;&gt;() {
            @Override
            public void call(Tuple2&lt;String, Iterable&lt;String&gt;&gt; tuple) throws Exception {
                String menpai = tuple._1;
                Iterator&lt;String&gt; iterator = tuple._2.iterator();
                String people = "";
                while (iterator.hasNext()){
                    people = people + iterator.next()+" ";
                }
                System.out.println("门派:"+menpai + "人员:"+people);
            }
        });

    }</pre>
</div>
<h4>（2）使用Java8编写</h4>
<div class="cnblogs_code">
<pre>public static void groupByKey(){
        List&lt;Tuple2&lt;String,String&gt;&gt; list = Arrays.asList(
                new Tuple2("武当", "张三丰"),
                new Tuple2("峨眉", "灭绝师太"),
                new Tuple2("武当", "宋青书"),
                new Tuple2("峨眉", "周芷若")
        );
        JavaPairRDD&lt;String, String&gt; listRDD = sc.parallelizePairs(list);

        JavaPairRDD&lt;String, Iterable&lt;String&gt;&gt; groupByKeyRDD = listRDD.groupByKey();
        groupByKeyRDD.foreach(tuple -&gt; {
            String menpai = tuple._1;
            Iterator&lt;String&gt; iterator = tuple._2.iterator();
            String people = "";
            while (iterator.hasNext()){
                people = people + iterator.next()+" ";
            }
            System.out.println("门派:"+menpai + "人员:"+people);
        });
    }</pre>
</div>
<h4>（3）使用scala编写</h4>
<div class="cnblogs_code">
<pre>def groupByKey(): Unit ={
    val list = List(("武当", "张三丰"), ("峨眉", "灭绝师太"), ("武当", "宋青书"), ("峨眉", "周芷若"))
    val listRDD = sc.parallelize(list)
    val groupByKeyRDD = listRDD.groupByKey()
    groupByKeyRDD.foreach(t =&gt; {
      val menpai = t._1
      val iterator = t._2.iterator
      var people = ""
      while (iterator.hasNext) people = people + iterator.next + " "
      println("门派:" + menpai + "人员:" + people)
    })
  }</pre>
</div>
<h4>（4）运行结果</h4>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201804/1228818-20180424193745000-2072522260.png" alt="" /></p>
<h3>3.3　join</h3>
<h4>（1）使用Java7编写</h4>
<p>join是将两个PairRDD合并，并将有相同key的元素分为一组，可以理解为groupByKey和Union的结合</p>
<div class="cnblogs_code">
<pre>public static void join(){
        final List&lt;Tuple2&lt;Integer, String&gt;&gt; names = Arrays.asList(
                new Tuple2&lt;Integer, String&gt;(1, "东方不败"),
                new Tuple2&lt;Integer, String&gt;(2, "令狐冲"),
                new Tuple2&lt;Integer, String&gt;(3, "林平之")
        );
        final List&lt;Tuple2&lt;Integer, Integer&gt;&gt; scores = Arrays.asList(
                new Tuple2&lt;Integer, Integer&gt;(1, 99),
                new Tuple2&lt;Integer, Integer&gt;(2, 98),
                new Tuple2&lt;Integer, Integer&gt;(3, 97)
        );

        final JavaPairRDD&lt;Integer, String&gt; nemesrdd = sc.parallelizePairs(names);
        final JavaPairRDD&lt;Integer, Integer&gt; scoresrdd = sc.parallelizePairs(scores);
        /**
         * &lt;Integer, 学号
         * Tuple2&lt;String, 名字
         * Integer&gt;&gt; 分数
         */
        final JavaPairRDD&lt;Integer, Tuple2&lt;String, Integer&gt;&gt; joinRDD = nemesrdd.join(scoresrdd);
//        final JavaPairRDD&lt;Integer, Tuple2&lt;Integer, String&gt;&gt; join = scoresrdd.join(nemesrdd);
        joinRDD.foreach(new VoidFunction&lt;Tuple2&lt;Integer, Tuple2&lt;String, Integer&gt;&gt;&gt;() {
            @Override
            public void call(Tuple2&lt;Integer, Tuple2&lt;String, Integer&gt;&gt; tuple) throws Exception {
                System.out.println("学号：" + tuple._1 + " 名字："+tuple._2._1 + " 分数："+tuple._2._2);
            }
        });
    }</pre>
</div>
<h4>（2）使用Java8编写</h4>
<div class="cnblogs_code">
<pre>public static void join(){
        final List&lt;Tuple2&lt;Integer, String&gt;&gt; names = Arrays.asList(
                new Tuple2&lt;Integer, String&gt;(1, "东方不败"),
                new Tuple2&lt;Integer, String&gt;(2, "令狐冲"),
                new Tuple2&lt;Integer, String&gt;(3, "林平之")
        );
        final List&lt;Tuple2&lt;Integer, Integer&gt;&gt; scores = Arrays.asList(
                new Tuple2&lt;Integer, Integer&gt;(1, 99),
                new Tuple2&lt;Integer, Integer&gt;(2, 98),
                new Tuple2&lt;Integer, Integer&gt;(3, 97)
        );

        final JavaPairRDD&lt;Integer, String&gt; nemesrdd = sc.parallelizePairs(names);
        final JavaPairRDD&lt;Integer, Integer&gt; scoresrdd = sc.parallelizePairs(scores);

        final JavaPairRDD&lt;Integer, Tuple2&lt;String, Integer&gt;&gt; joinRDD = nemesrdd.join(scoresrdd);
        joinRDD.foreach(tuple -&gt; System.out.println("学号:"+tuple._1+" 姓名:"+tuple._2._1+" 成绩:"+tuple._2._2));
    }</pre>
</div>
<h4>（3）使用scala编写</h4>
<div class="cnblogs_code">
<pre>def join(): Unit = {
    val list1 = List((1, "东方不败"), (2, "令狐冲"), (3, "林平之"))
    val list2 = List((1, 99), (2, 98), (3, 97))
    val list1RDD = sc.parallelize(list1)
    val list2RDD = sc.parallelize(list2)

    val joinRDD = list1RDD.join(list2RDD)
    joinRDD.foreach(t =&gt; println("学号:" + t._1 + " 姓名:" + t._2._1 + " 成绩:" + t._2._2))

  }</pre>
</div>
<h4>（4）运行结果</h4>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201804/1228818-20180424195058477-646994908.png" alt="" /></p>
<h2>四、sample、cartesian&nbsp;</h2>
<h3>4.1　sample</h3>
<p>（1）使用Java7编写</p>
<div class="cnblogs_code">
<pre> public static void sample(){
        ArrayList&lt;Integer&gt; list = new ArrayList&lt;&gt;();
        for(int i=1;i&lt;=100;i++){
            list.add(i);
        }
        JavaRDD&lt;Integer&gt; listRDD = sc.parallelize(list);
        /**
         * sample用来从RDD中抽取样本。他有三个参数
         * withReplacement: Boolean,
         *       true: 有放回的抽样
         *       false: 无放回抽象
         * fraction: Double：
         *      抽取样本的比例
         * seed: Long：
         *      随机种子
         */
        JavaRDD&lt;Integer&gt; sampleRDD = listRDD.sample(false, 0.1,0);
        sampleRDD.foreach(new VoidFunction&lt;Integer&gt;() {
            @Override
            public void call(Integer num) throws Exception {
                System.out.print(num+" ");
            }
        });
    }</pre>
</div>
<h4>（2）使用Java8编写</h4>
<div class="cnblogs_code">
<pre>public static void sample(){
        ArrayList&lt;Integer&gt; list = new ArrayList&lt;&gt;();
        for(int i=1;i&lt;=100;i++){
            list.add(i);
        }
        JavaRDD&lt;Integer&gt; listRDD = sc.parallelize(list);

        JavaRDD&lt;Integer&gt; sampleRDD = listRDD.sample(false, 0.1, 0);
        sampleRDD.foreach(num -&gt; System.out.print(num + " "));
    }</pre>
</div>
<h4>（3）使用scala编写</h4>
<div class="cnblogs_code">
<pre>  def sample(): Unit ={
    val list = 1 to 100
    val listRDD = sc.parallelize(list)
    listRDD.sample(false,0.1,0).foreach(num =&gt; print(num + " "))
  }</pre>
</div>
<h4>（4）运行结果</h4>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201804/1228818-20180424202609449-616825750.png" alt="" /></p>
<h3>4.2　cartesian</h3>
<p>cartesian是用于求笛卡尔积的</p>
<h4>（1）使用Java7编写</h4>
<div class="cnblogs_code">
<pre>public static void cartesian(){
        List&lt;String&gt; list1 = Arrays.asList("A", "B");
        List&lt;Integer&gt; list2 = Arrays.asList(1, 2, 3);
        JavaRDD&lt;String&gt; list1RDD = sc.parallelize(list1);
        JavaRDD&lt;Integer&gt; list2RDD = sc.parallelize(list2);
        list1RDD.cartesian(list2RDD).foreach(new VoidFunction&lt;Tuple2&lt;String, Integer&gt;&gt;() {
            @Override
            public void call(Tuple2&lt;String, Integer&gt; tuple) throws Exception {
                System.out.println(tuple._1 + "-&gt;" + tuple._2);
            }
        });

    }</pre>
</div>
<h4>（2）使用Java8编写</h4>
<div class="cnblogs_code">
<pre> public static void cartesian(){
        List&lt;String&gt; list1 = Arrays.asList("A", "B");
        List&lt;Integer&gt; list2 = Arrays.asList(1, 2, 3);
        JavaRDD&lt;String&gt; list1RDD = sc.parallelize(list1);
        JavaRDD&lt;Integer&gt; list2RDD = sc.parallelize(list2);
        list1RDD.cartesian(list2RDD).foreach(tuple -&gt; System.out.print(tuple._1 + "-&gt;" + tuple._2));
    }</pre>
</div>
<h4>（3）使用scala编写</h4>
<div class="cnblogs_code">
<pre>def cartesian(): Unit ={
    val list1 = List("A","B")
    val list2 = List(1,2,3)
    val list1RDD = sc.parallelize(list1)
    val list2RDD = sc.parallelize(list2)
    list1RDD.cartesian(list2RDD).foreach(t =&gt; println(t._1 +"-&gt;"+t._2))
  }</pre>
</div>
<h4>（4）运行结果</h4>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201804/1228818-20180424203611915-24763542.png" alt="" /></p>
<h2>五、filter、distinct、intersection</h2>
<h3>5.1　filter</h3>
<h4>（1）使用Java7编写</h4>
<p>过滤出偶数</p>
<div class="cnblogs_code">
<pre>public static void filter(){
        List&lt;Integer&gt; list = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
        JavaRDD&lt;Integer&gt; listRDD = sc.parallelize(list);
        JavaRDD&lt;Integer&gt; filterRDD = listRDD.filter(new Function&lt;Integer, Boolean&gt;() {
            @Override
            public Boolean call(Integer num) throws Exception {
                return num % 2 == 0;
            }
        });
        filterRDD.foreach(new VoidFunction&lt;Integer&gt;() {
            @Override
            public void call(Integer num) throws Exception {
                System.out.print(num + " ");
            }
        });

    }</pre>
</div>
<h4>（2）使用Java8编写</h4>
<div class="cnblogs_code">
<pre>    public static void filter(){
        List&lt;Integer&gt; list = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
        JavaRDD&lt;Integer&gt; listRDD = sc.parallelize(list);
        JavaRDD&lt;Integer&gt; filterRDD = listRDD.filter(num -&gt; num % 2 ==0);
        filterRDD.foreach(num -&gt; System.out.print(num + " "));
    }</pre>
</div>
<h4>（3）使用scala编写</h4>
<div class="cnblogs_code">
<pre>  def filter(): Unit ={
    val list = List(1,2,3,4,5,6,7,8,9,10)
    val listRDD = sc.parallelize(list)
    listRDD.filter(num =&gt; num % 2 ==0).foreach(print(_))
  }</pre>
</div>
<h4>（4）运行结果</h4>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201804/1228818-20180424205241479-1065453041.png" alt="" /></p>
<h3>5.2　distinct</h3>
<h4>（1）使用Java7编写</h4>
<div class="cnblogs_code">
<pre>public static void distinct(){
        List&lt;Integer&gt; list = Arrays.asList(1, 1, 2, 2, 3, 3, 4, 5);
        JavaRDD&lt;Integer&gt; listRDD  = (JavaRDD&lt;Integer&gt;) sc.parallelize(list);
        JavaRDD&lt;Integer&gt; distinctRDD = listRDD.distinct();
        distinctRDD.foreach(new VoidFunction&lt;Integer&gt;() {
            @Override
            public void call(Integer num) throws Exception {
                System.out.println(num);
            }
        });
    }</pre>
</div>
<h4>（2）使用Java8编写</h4>
<div class="cnblogs_code">
<pre> public static void distinct(){
        List&lt;Integer&gt; list = Arrays.asList(1, 1, 2, 2, 3, 3, 4, 5);
        JavaRDD&lt;Integer&gt; listRDD  = (JavaRDD&lt;Integer&gt;) sc.parallelize(list);
        listRDD.distinct().foreach(num -&gt; System.out.println(num));
    }</pre>
</div>
<h4>（3）使用scala编写</h4>
<div class="cnblogs_code">
<pre> def distinct(): Unit ={
    val list = List(1,1,2,2,3,3,4,5)
    sc.parallelize(list).distinct().foreach(println(_))
  }</pre>
</div>
<h4>（4）运行结果</h4>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201804/1228818-20180424205936796-277394349.png" alt="" /></p>
<h3>5.3　intersection</h3>
<h4>（1）使用Java7编写</h4>
<div class="cnblogs_code">
<pre>public static void intersection(){
        List&lt;Integer&gt; list1 = Arrays.asList(1, 2, 3, 4);
        List&lt;Integer&gt; list2 = Arrays.asList(3, 4, 5, 6);
        JavaRDD&lt;Integer&gt; list1RDD = sc.parallelize(list1);
        JavaRDD&lt;Integer&gt; list2RDD = sc.parallelize(list2);
        list1RDD.intersection(list2RDD).foreach(new VoidFunction&lt;Integer&gt;() {
            @Override
            public void call(Integer num) throws Exception {
                System.out.println(num);
            }
        });
    }</pre>
</div>
<h4>（2）使用Java8编写</h4>
<div class="cnblogs_code">
<pre> public static void intersection() {
        List&lt;Integer&gt; list1 = Arrays.asList(1, 2, 3, 4);
        List&lt;Integer&gt; list2 = Arrays.asList(3, 4, 5, 6);
        JavaRDD&lt;Integer&gt; list1RDD = sc.parallelize(list1);
        JavaRDD&lt;Integer&gt; list2RDD = sc.parallelize(list2);
        list1RDD.intersection(list2RDD).foreach(num -&gt;System.out.println(num));
    }</pre>
</div>
<h4>（3）使用scala编写</h4>
<div class="cnblogs_code">
<pre>def intersection(): Unit ={
    val list1 = List(1,2,3,4)
    val list2 = List(3,4,5,6)
    val list1RDD = sc.parallelize(list1)
    val list2RDD = sc.parallelize(list2)
    list1RDD.intersection(list2RDD).foreach(println(_))
  }</pre>
</div>
<h4>（4）运行结果</h4>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201804/1228818-20180425095821651-1951412641.png" alt="" /></p>
<h2>六、coalesce、repartition、repartitionAndSortWithinPartitions</h2>
<h3>6.1　coalesce</h3>
<p>分区数由多&nbsp; -》 变少</p>
<h4>（1）使用Java7编写</h4>
<div class="cnblogs_code">
<pre>public static void coalesce(){
        List&lt;Integer&gt; list = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9);
        JavaRDD&lt;Integer&gt; listRDD = sc.parallelize(list, 3);
        listRDD.coalesce(1).foreach(new VoidFunction&lt;Integer&gt;() {
            @Override
            public void call(Integer num) throws Exception {
                System.out.print(num);
            }
        });
    }</pre>
</div>
<h4>（2）使用Java8编写</h4>
<div class="cnblogs_code">
<pre>public static void coalesce() {
        List&lt;Integer&gt; list = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9);
        JavaRDD&lt;Integer&gt; listRDD = sc.parallelize(list, 3);
        listRDD.coalesce(1).foreach(num -&gt; System.out.println(num));
    }</pre>
</div>
<h4>（3）使用scala编写</h4>
<div class="cnblogs_code">
<pre>def coalesce(): Unit = {
    val list = List(1,2,3,4,5,6,7,8,9)
    sc.parallelize(list,3).coalesce(1).foreach(println(_))
  }</pre>
</div>
<h4>（4）运行结果</h4>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201804/1228818-20180425100626329-329630922.png" alt="" /></p>
<h3>6.2 replication</h3>
<p>&nbsp;进行重分区，解决的问题：本来分区数少&nbsp; -》 增加分区数</p>
<h4>（1）使用Java7编写</h4>
<div class="cnblogs_code">
<pre> public static void replication(){
        List&lt;Integer&gt; list = Arrays.asList(1, 2, 3, 4);
        JavaRDD&lt;Integer&gt; listRDD = sc.parallelize(list, 1);
        listRDD.repartition(2).foreach(new VoidFunction&lt;Integer&gt;() {
            @Override
            public void call(Integer num) throws Exception {
                System.out.println(num);
            }
        });
    }</pre>
</div>
<h4>（2）使用Java8编写</h4>
<div class="cnblogs_code">
<pre>public static void replication(){
        List&lt;Integer&gt; list = Arrays.asList(1, 2, 3, 4);
        JavaRDD&lt;Integer&gt; listRDD = sc.parallelize(list, 1);
        listRDD.repartition(2).foreach(num -&gt; System.out.println(num));
    }</pre>
</div>
<h4>（3）使用scala编写</h4>
<div class="cnblogs_code">
<pre>def replication(): Unit ={
    val list = List(1,2,3,4)
    val listRDD = sc.parallelize(list,1)
    listRDD.repartition(2).foreach(println(_))
  }</pre>
</div>
<h4>（4）运行结果</h4>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201804/1228818-20180425101846994-1633998903.png" alt="" /></p>
<h3>6.3　repartitionAndSortWithinPartitions</h3>
<p>repartitionAndSortWithinPartitions函数是repartition函数的变种，与repartition函数不同的是，repartitionAndSortWithinPartitions在给定的partitioner内部进行排序，性能比repartition要高。&nbsp;</p>
<h4>（1）使用Java7编写</h4>
<div class="cnblogs_code">
<pre>public static void repartitionAndSortWithinPartitions(){
        List&lt;Integer&gt; list = Arrays.asList(1, 3, 55, 77, 33, 5, 23);
        JavaRDD&lt;Integer&gt; listRDD = sc.parallelize(list, 1);
        JavaPairRDD&lt;Integer, Integer&gt; pairRDD = listRDD.mapToPair(new PairFunction&lt;Integer, Integer, Integer&gt;() {
            @Override
            public Tuple2&lt;Integer, Integer&gt; call(Integer num) throws Exception {
                return new Tuple2&lt;&gt;(num, num);
            }
        });
        JavaPairRDD&lt;Integer, Integer&gt; parationRDD = pairRDD.repartitionAndSortWithinPartitions(new Partitioner() {
            @Override
            public int getPartition(Object key) {
                Integer index = Integer.valueOf(key.toString());
                if (index % 2 == 0) {
                    return 0;
                } else {
                    return 1;
                }

            }

            @Override
            public int numPartitions() {
                return 2;
            }
        });
        parationRDD.mapPartitionsWithIndex(new Function2&lt;Integer, Iterator&lt;Tuple2&lt;Integer, Integer&gt;&gt;, Iterator&lt;String&gt;&gt;() {
            @Override
            public  Iterator&lt;String&gt; call(Integer index, Iterator&lt;Tuple2&lt;Integer, Integer&gt;&gt; iterator) throws Exception {
                final ArrayList&lt;String&gt; list1 = new ArrayList&lt;&gt;();
                while (iterator.hasNext()){
                    list1.add(index+"_"+iterator.next());
                }
                return list1.iterator();
            }
        },false).foreach(new VoidFunction&lt;String&gt;() {
            @Override
            public void call(String s) throws Exception {
                System.out.println(s);
            }
        });
    }</pre>
</div>
<h4>（2）使用Java8编写</h4>
<div class="cnblogs_code">
<pre>public static void repartitionAndSortWithinPartitions(){
        List&lt;Integer&gt; list = Arrays.asList(1, 4, 55, 66, 33, 48, 23);
        JavaRDD&lt;Integer&gt; listRDD = sc.parallelize(list, 1);
        JavaPairRDD&lt;Integer, Integer&gt; pairRDD = listRDD.mapToPair(num -&gt; new Tuple2&lt;&gt;(num, num));
        pairRDD.repartitionAndSortWithinPartitions(new HashPartitioner(2))
                .mapPartitionsWithIndex((index,iterator) -&gt; {
                    ArrayList&lt;String&gt; list1 = new ArrayList&lt;&gt;();
                    while (iterator.hasNext()){
                        list1.add(index+"_"+iterator.next());
                    }
                    return list1.iterator();
                },false)
                .foreach(str -&gt; System.out.println(str));
    }</pre>
</div>
<h4>（3）使用scala编写</h4>
<div class="cnblogs_code">
<pre>def repartitionAndSortWithinPartitions(): Unit ={
    val list = List(1, 4, 55, 66, 33, 48, 23)
    val listRDD = sc.parallelize(list,1)
    listRDD.map(num =&gt; (num,num))
      .repartitionAndSortWithinPartitions(new HashPartitioner(2))
      .mapPartitionsWithIndex((index,iterator) =&gt; {
        val listBuffer: ListBuffer[String] = new ListBuffer
        while (iterator.hasNext) {
          listBuffer.append(index + "_" + iterator.next())
        }
        listBuffer.iterator
      },false)
      .foreach(println(_))

  }</pre>
</div>
<h4>（4）运行结果</h4>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201804/1228818-20180425111645221-315536430.png" alt="" height="180" /><img src="https://images2018.cnblogs.com/blog/1228818/201804/1228818-20180425111716570-1751515138.png" alt="" height="180" /></p>
<h2>七、cogroup、sortBykey、aggregateByKey</h2>
<h3>7.1　cogroup</h3>
<p>对两个RDD中的KV元素，每个RDD中相同key中的元素分别聚合成一个集合。与reduceByKey不同的是针对两个RDD中相同的key的元素进行合并。</p>
<h4>（1）使用Java7编写</h4>
<div class="cnblogs_code">
<pre>public static void cogroup(){
        List&lt;Tuple2&lt;Integer, String&gt;&gt; list1 = Arrays.asList(
                new Tuple2&lt;Integer, String&gt;(1, "www"),
                new Tuple2&lt;Integer, String&gt;(2, "bbs")
        );

        List&lt;Tuple2&lt;Integer, String&gt;&gt; list2 = Arrays.asList(
                new Tuple2&lt;Integer, String&gt;(1, "cnblog"),
                new Tuple2&lt;Integer, String&gt;(2, "cnblog"),
                new Tuple2&lt;Integer, String&gt;(3, "very")
        );

        List&lt;Tuple2&lt;Integer, String&gt;&gt; list3 = Arrays.asList(
                new Tuple2&lt;Integer, String&gt;(1, "com"),
                new Tuple2&lt;Integer, String&gt;(2, "com"),
                new Tuple2&lt;Integer, String&gt;(3, "good")
        );

        JavaPairRDD&lt;Integer, String&gt; list1RDD = sc.parallelizePairs(list1);
        JavaPairRDD&lt;Integer, String&gt; list2RDD = sc.parallelizePairs(list2);
        JavaPairRDD&lt;Integer, String&gt; list3RDD = sc.parallelizePairs(list3);

        list1RDD.cogroup(list2RDD,list3RDD).foreach(new VoidFunction&lt;Tuple2&lt;Integer, Tuple3&lt;Iterable&lt;String&gt;, Iterable&lt;String&gt;, Iterable&lt;String&gt;&gt;&gt;&gt;() {
            @Override
            public void call(Tuple2&lt;Integer, Tuple3&lt;Iterable&lt;String&gt;, Iterable&lt;String&gt;, Iterable&lt;String&gt;&gt;&gt; tuple) throws Exception {
                System.out.println(tuple._1+" " +tuple._2._1() +" "+tuple._2._2()+" "+tuple._2._3());
            }
        });
    }</pre>
</div>
<h4>（2）使用Java8编写</h4>
<div class="cnblogs_code">
<pre>public static void cogroup(){
        List&lt;Tuple2&lt;Integer, String&gt;&gt; list1 = Arrays.asList(
                new Tuple2&lt;Integer, String&gt;(1, "www"),
                new Tuple2&lt;Integer, String&gt;(2, "bbs")
        );

        List&lt;Tuple2&lt;Integer, String&gt;&gt; list2 = Arrays.asList(
                new Tuple2&lt;Integer, String&gt;(1, "cnblog"),
                new Tuple2&lt;Integer, String&gt;(2, "cnblog"),
                new Tuple2&lt;Integer, String&gt;(3, "very")
        );

        List&lt;Tuple2&lt;Integer, String&gt;&gt; list3 = Arrays.asList(
                new Tuple2&lt;Integer, String&gt;(1, "com"),
                new Tuple2&lt;Integer, String&gt;(2, "com"),
                new Tuple2&lt;Integer, String&gt;(3, "good")
        );

        JavaPairRDD&lt;Integer, String&gt; list1RDD = sc.parallelizePairs(list1);
        JavaPairRDD&lt;Integer, String&gt; list2RDD = sc.parallelizePairs(list2);
        JavaPairRDD&lt;Integer, String&gt; list3RDD = sc.parallelizePairs(list3);

        list1RDD.cogroup(list2RDD,list3RDD).foreach(tuple -&gt;
                System.out.println(tuple._1+" " +tuple._2._1() +" "+tuple._2._2()+" "+tuple._2._3()));
    }</pre>
</div>
<h4>（3）使用scala编写</h4>
<div class="cnblogs_code">
<pre>def cogroup(): Unit ={
    val list1 = List((1, "www"), (2, "bbs"))
    val list2 = List((1, "cnblog"), (2, "cnblog"), (3, "very"))
    val list3 = List((1, "com"), (2, "com"), (3, "good"))

    val list1RDD = sc.parallelize(list1)
    val list2RDD = sc.parallelize(list2)
    val list3RDD = sc.parallelize(list3)

    list1RDD.cogroup(list2RDD,list3RDD).foreach(tuple =&gt;
      println(tuple._1 + " " + tuple._2._1 + " " + tuple._2._2 + " " + tuple._2._3))
  }</pre>
</div>
<h4>（4）运行结果</h4>
<p><img src="https://images2018.cnblogs.com/blog/1228818/201804/1228818-20180425141346251-899700830.png" alt="" /></p>
<h3>7.2　sortBykey</h3>
<p>sortByKey函数作用于Key-Value形式的RDD，并对Key进行排序。它是在<code>org.apache.spark.rdd.OrderedRDDFunctions</code>中实现的，实现如下</p>
<div class="cnblogs_code">
<pre>def sortByKey(ascending: Boolean = true, numPartitions: Int = self.partitions.size)
    : RDD[(K, V)] =
{
  val part = new RangePartitioner(numPartitions, self, ascending)
  new ShuffledRDD[K, V, V](self, part)
    .setKeyOrdering(if (ascending) ordering else ordering.reverse)
}</pre>
</div>
<p>从函数的实现可以看出，它主要接受两个函数，含义和sortBy一样，这里就不进行解释了。该函数返回的RDD一定是ShuffledRDD类型的，因为对源RDD进行排序，必须进行Shuffle操作，而Shuffle操作的结果RDD就是ShuffledRDD。其实这个函数的实现很优雅，里面用到了RangePartitioner，它可以使得相应的范围Key数据分到同一个partition中，然后内部用到了mapPartitions对每个partition中的数据进行排序，而每个partition中数据的排序用到了标准的sort机制，避免了大量数据的shuffle。下面对sortByKey的使用进行说明：</p>
<h4>（1）使用Java7编写</h4>
<div class="cnblogs_code">
<pre>public static void sortByKey(){
        List&lt;Tuple2&lt;Integer, String&gt;&gt; list = Arrays.asList(
                new Tuple2&lt;&gt;(99, "张三丰"),
                new Tuple2&lt;&gt;(96, "东方不败"),
                new Tuple2&lt;&gt;(66, "林平之"),
                new Tuple2&lt;&gt;(98, "聂风")
        );
        JavaPairRDD&lt;Integer, String&gt; listRDD = sc.parallelizePairs(list);
        listRDD.sortByKey(false).foreach(new VoidFunction&lt;Tuple2&lt;Integer, String&gt;&gt;() {
            @Override
            public void call(Tuple2&lt;Integer, String&gt; tuple) throws Exception {
                System.out.println(tuple._2+"-&gt;"+tuple._1);
            }
        });
    }</pre>
</div>
<h4>（2）使用Java8编写</h4>
<div class="cnblogs_code">
<pre>public static void sortByKey(){
        List&lt;Tuple2&lt;Integer, String&gt;&gt; list = Arrays.asList(
                new Tuple2&lt;&gt;(99, "张三丰"),
                new Tuple2&lt;&gt;(96, "东方不败"),
                new Tuple2&lt;&gt;(66, "林平之"),
                new Tuple2&lt;&gt;(98, "聂风")
        );
        JavaPairRDD&lt;Integer, String&gt; listRDD = sc.parallelizePairs(list);
        listRDD.sortByKey(false).foreach(tuple -&gt;System.out.println(tuple._2+"-&gt;"+tuple._1));
    }</pre>
</div>
<h4>（3）使用scala编写</h4>
<div class="cnblogs_code">
<pre>def sortByKey(): Unit ={
    val list = List((99, "张三丰"), (96, "东方不败"), (66, "林平之"), (98, "聂风"))
    sc.parallelize(list).sortByKey(false).foreach(tuple =&gt; println(tuple._2 + "-&gt;" + tuple._1))
  }</pre>
</div>
<h4>（4）运行结果</h4>
<p>&nbsp;<img src="https://images2018.cnblogs.com/blog/1228818/201804/1228818-20180425142713725-2107309925.png" alt="" /></p>
<h3>7.3　aggregateByKey</h3>
<p>aggregateByKey函数对PairRDD中相同Key的值进行聚合操作，在聚合过程中同样使用了一个中立的初始值。和aggregate函数类似，aggregateByKey返回值的类型不需要和RDD中value的类型一致。因为aggregateByKey是对相同Key中的值进行聚合操作，所以aggregateByKey函数最终返回的类型还是Pair RDD，对应的结果是Key和聚合好的值；而aggregate函数直接是返回非RDD的结果，这点需要注意。在实现过程中，定义了三个aggregateByKey函数原型，但最终调用的aggregateByKey函数都一致。</p>
<h4>（1）使用Java7编写</h4>
<div class="cnblogs_code">
<pre>public static void aggregateByKey(){
        List&lt;String&gt; list = Arrays.asList("you,jump", "i,jump");
        JavaRDD&lt;String&gt; listRDD = sc.parallelize(list);
        listRDD.flatMap(new FlatMapFunction&lt;String, String&gt;() {
            @Override
            public Iterator&lt;String&gt; call(String line) throws Exception {
                return Arrays.asList(line.split(",")).iterator();
            }
        }).mapToPair(new PairFunction&lt;String, String, Integer&gt;() {
            @Override
            public Tuple2&lt;String, Integer&gt; call(String word) throws Exception {
                return new Tuple2&lt;&gt;(word,1);
            }
        }).aggregateByKey(0, new Function2&lt;Integer, Integer, Integer&gt;() {
            @Override
            public Integer call(Integer i1, Integer i2) throws Exception {
                return i1 + i2;
            }
        }, new Function2&lt;Integer, Integer, Integer&gt;() {
            @Override
            public Integer call(Integer i1, Integer i2) throws Exception {
                return i1+i2;
            }
        }).foreach(new VoidFunction&lt;Tuple2&lt;String, Integer&gt;&gt;() {
            @Override
            public void call(Tuple2&lt;String, Integer&gt; tuple) throws Exception {
                System.out.println(tuple._1+"-&gt;"+tuple._2);
            }
        });
    }</pre>
</div>
<h4>（2）使用Java8编写</h4>
<div class="cnblogs_code">
<pre>public static void aggregateByKey() {
        List&lt;String&gt; list = Arrays.asList("you,jump", "i,jump");
        JavaRDD&lt;String&gt; listRDD = sc.parallelize(list);
        listRDD.flatMap(line -&gt; Arrays.asList(line.split(",")).iterator())
                .mapToPair(word -&gt; new Tuple2&lt;&gt;(word,1))
                .aggregateByKey(0,(x,y)-&gt; x+y,(m,n) -&gt; m+n)
                .foreach(tuple -&gt; System.out.println(tuple._1+"-&gt;"+tuple._2));
    }</pre>
</div>
<h4>（3）使用scala编写</h4>
<div class="cnblogs_code">
<pre>def aggregateByKey(): Unit ={
    val list = List("you,jump", "i,jump")
    sc.parallelize(list)
      .flatMap(_.split(","))
      .map((_, 1))
      .aggregateByKey(0)(_+_,_+_)
      .foreach(tuple =&gt;println(tuple._1+"-&gt;"+tuple._2))
  }</pre>
</div>
<h4>（4）运行结果</h4>
<p>&nbsp;<img src="https://images2018.cnblogs.com/blog/1228818/201804/1228818-20180425145645089-1948052432.png" alt="" /></p></div>

</body>
</html>
